/*
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.apache.pig.pigpen.editor.parser;

import java.io.ByteArrayInputStream;
import java.io.File;
import java.io.IOException;
import java.io.InputStream;
import java.io.Reader;
import java.util.HashMap;
import java.util.HashSet;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Properties;

import org.apache.hadoop.conf.Configuration;
import org.apache.pig.ExecType;
import org.apache.pig.PigServer;
import org.apache.pig.backend.hadoop.datastorage.ConfigurationUtil;
import org.apache.pig.data.DataBag;
import org.apache.pig.impl.PigContext;
import org.apache.pig.impl.io.FileSpec;
import org.apache.pig.impl.logicalLayer.FrontendException;
import org.apache.pig.impl.logicalLayer.LOLoad;
import org.apache.pig.impl.logicalLayer.LOStore;
import org.apache.pig.impl.logicalLayer.LogicalOperator;
import org.apache.pig.impl.logicalLayer.LogicalPlan;
import org.apache.pig.impl.logicalLayer.LogicalPlanBuilder;
import org.apache.pig.impl.logicalLayer.parser.ParseException;
import org.apache.pig.impl.logicalLayer.schema.Schema;
import org.apache.pig.impl.plan.OperatorKey;
import org.apache.pig.pigpen.ConsoleLogger;
import org.apache.pig.pigpen.PigPenActivator;
import org.apache.pig.pigpen.LoggerUtil;
import org.apache.pig.pigpen.PluginConfiguration;
import org.apache.pig.pigpen.editor.model.Link;
import org.apache.pig.pigpen.editor.model.Operator;
import org.apache.pig.pigpen.editor.model.Query;
import org.apache.pig.pigpen.preferences.PreferenceConstants;
import org.apache.pig.pigpen.util.PigServerPool;
import org.apache.pig.tools.pigscript.parser.PigScriptParser;
import org.apache.pig.tools.pigscript.parser.PigScriptParserTokenManager;
import org.eclipse.jface.preference.IPreferenceStore;

public class PigPenParser extends PigScriptParser {

    private Query query;
    private boolean mDone;

    // pig parser requires a "pigContext" object
    private PigContext pigContext;

    // aliases parsed
    // private Map<String, IntermedResult> aliasToIR;
    private Map<String, LogicalPlan> aliasToPlan;
    private Map<LogicalOperator, LogicalPlan> aliases;
    private Map<String, LogicalOperator> aliasOp;
    // private Map<String, LogicalOperator> aliasToLogOp;
    // private Map<String, Operator> aliasToOp;
    private Map<OperatorKey, LogicalOperator> opTable;

    private Map<Operator, LogicalOperator> opToLogOp;
    private Map<LogicalOperator, Operator> LogOpToOp;
    private Map<String, String> fileNameMap;
    private ConsoleLogger logger = ConsoleLogger.getDefault();
    // whether there were one or more duplicate aliases
    boolean dupAliasAttempt;

    // the operators that we need to reparse
    private List<Operator> toReparse;

    // the opeartor we're currently reparsing (null if we're parsing a query
    // from scratch)
    private Operator reParseOp;

    private PigServer pigServer;

    private static final String JOB_TRACKER_LOCATION = "mapred.job.tracker";
    private static final String FILE_SYSTEM_LOCATION = "fs.default.name";

    public boolean PigServerStart = false;

    public PigPenParser() {
        super(new ByteArrayInputStream("\n".getBytes()));
        init();
    }

    public PigPenParser(Reader stream) {
        super(stream);
        init();
    }

    public PigPenParser(InputStream stream, String encoding) {
        super(stream, encoding);
        init();
    }

    public PigPenParser(InputStream stream) {
        super(stream);
        init();
    }

    public PigPenParser(PigScriptParserTokenManager tm) {
        super(tm);
        init();
    }

    private void init() {

        aliasToPlan = new HashMap<String, LogicalPlan>();
        toReparse = new LinkedList<Operator>();
        opTable = new HashMap<OperatorKey, LogicalOperator>();

        opToLogOp = new HashMap<Operator, LogicalOperator>();
        LogOpToOp = new HashMap<LogicalOperator, Operator>();

        aliases = new HashMap<LogicalOperator, LogicalPlan>();
        aliasOp = new HashMap<String, LogicalOperator>();
        fileNameMap = new HashMap<String, String>();
        //PigServerStart = PigServerPool.getPigServer();//initPigServer();
        pigServer = PigServerPool.getPigServer();
    }

//    private boolean initPigServer() {
//        IPreferenceStore store = PigPenActivator.getDefault()
//                .getPreferenceStore();
//        String stringList = store.getString(PreferenceConstants.P_PROPERTIES);
//        Map<String, String> PrefProperties = PluginConfiguration
//                .parseString(stringList);
//        String confPath = PrefProperties.get(PreferenceConstants.P_CONFPATH);
//
//        if (confPath == null) {
//            logger.clear();
//            logger.setFocus();
//            logger
//                    .error("Please set the ConfigurationPath variable in Pig Preferences (Window -> Preferences -> Pig) to point to the directory containing hadoop-site.xml and pig.properties");
//            throw new RuntimeException("Fatal error!!!!");
//            // return false;
//        }
//        logger.clear();
//        if (confPath.charAt(confPath.length() - 1) != '/')
//            confPath = confPath + "/";
//
//        try {
//
//            Configuration conf = new Configuration();
//
//            conf.addResource((new File(confPath + "hadoop-site.xml")).toURI()
//                    .toURL());
//            // conf.addResource((new File(confPath +
//            // "pig.properties")).toURI().toURL().toString());
//            Properties properties = ConfigurationUtil.toProperties(conf);
//            pigServer = new PigServer(ExecType.MAPREDUCE, properties);
//            // pigServer = new PigServer(ExecType.MAPREDUCE, new Properties());
//
//            //System.out.println(pigContext.getDfs().getStatistics().toString())
//            // ;
//        } catch (Throwable e) {
//            logger.setFocus();
//            logger.error(e.getMessage());
//            logger
//                    .error("Please set the ConfigurationPath variable in Pig Preferences (Window -> Preferences -> Pig) to point to the directory containing hadoop-site.xml and pig.properties");
//        }
//        pigContext = pigServer.getPigContext();
//        logger.setFocus();
//        logger.info("Connected to map-reduce job tracker at: "
//                + pigContext.getProperties().getProperty(JOB_TRACKER_LOCATION));
//        logger.info("Connected to hadoop file system at: "
//                + pigContext.getProperties().getProperty(FILE_SYSTEM_LOCATION));
//        return true;
//    }

    // returns true iff a duplicate alias was detected
    public boolean parseQuery() {
        dupAliasAttempt = false;

        reParseOp = null;

        mDone = false;
        while (!mDone) {
            try {
                parse();
            } catch (Throwable e) {
                LoggerUtil
                        .logInfo("Encountered unparsable pig script command; skipping");
                logger
                        .error("Encountered unparsable pig script command; skipping\n"
                                + e.getMessage());
            }
        }

        return dupAliasAttempt;
    }

    public void reParseOp(Operator op) {
        dupAliasAttempt = false;

        toReparse.clear();
        toReparse.add(op);
        List<Operator> danglingOps = query.getDanglingOperators();
        danglingOps.remove(op);
        toReparse.addAll(danglingOps);

        // keep reparsing operators until unable to make any progress
        while (true) {
            // move reparse queue to private list
            List<Operator> reparseList = new LinkedList<Operator>(toReparse);
            toReparse.clear();

            for (Operator reparseMe : reparseList) {
                reParseOp = reparseMe;

                String cmd = reParseOp.getText();
                if (cmd.charAt(cmd.length() - 1) != ';')
                    cmd = cmd + ";";

                InputStream stream = new ByteArrayInputStream((cmd + "\n")
                        .getBytes());
                ReInit(stream);

                try {
                    parse();
                } catch (Throwable e) {
                    List<Link> outputs = new LinkedList<Link>(reParseOp
                            .getOutputs());

                    reParseOp.resetDerivedFields();
                    if (cmd.toLowerCase().startsWith("store")) // a hack, but it
                                                               // works
                        reParseOp.setName("Store");
                    reParseOp.removeAllInputs();
                    reParseOp.removeAllOutputs();

                    reParseOp.setError(e.getMessage());

                    for (Link output : outputs)
                        toReparse.add(output.getTarget());

                }

                try {
                    stream.close();
                } catch (IOException e) {
                    // just ignore
                }
            }

            // check whether we made any progress this round
            // (i.e., whether toReparse differs from reparseList)
            if (toReparse.size() == reparseList.size()) {
                boolean listsMatch = (new HashSet<Operator>(toReparse))
                        .equals(new HashSet<Operator>(reparseList));
                if (listsMatch)
                    break;
            }
        }
    }

    public void removeOpState(Operator op) {
        String alias = op.getAlias();

        if (alias != null) {
            // aliasToIR.remove(alias);
            aliasToPlan.remove(alias);
            /*
             * aliasToLogOp.remove(alias); aliasToOp.remove(alias);
             */
            LogicalOperator lop = opToLogOp.get(op);
            opToLogOp.remove(op);
            LogOpToOp.remove(lop);
        }
    }

    public void setQuery(Query query) {
        this.query = query;
    }

    protected void processCat(String path) throws IOException {
        // no-op
        LoggerUtil.logInfo("Ignoring cat command.");
    }

    protected void processDump(String alias) throws IOException {
        // no-op
        LoggerUtil.logInfo("Ignoring dump command.");
    }

    protected void processKill(String jobid) throws IOException {
        // no-op
        LoggerUtil.logInfo("Ignoring kill command.");
    }

    protected void processLS(String path) throws IOException {
        // no-op
        LoggerUtil.logInfo("Ignoring ls command.");
    }

    protected void printHelp() {
        // no-op
        LoggerUtil.logInfo("Ignoring help command.");
    }

    protected void processMove(String src, String dst) throws IOException {
        // no-op
        LoggerUtil.logInfo("Ignoring mv command.");

    }

    protected void processPig(String cmd) throws IOException {
        // add semicolon, if not already there
        if (cmd.charAt(cmd.length() - 1) != ';')
            cmd = cmd + ";";

        // operator info to populate
        String opName = Operator.DefaultName;
        String error = null;
        String exdata = null;
        String alias = null;
        Schema schema = null;

        // parse the query into a logical plan
        LogicalPlan lp = null;
        LogicalOperator lo = null;

        try {
            // lp = ((new
            // LogicalPlanBuilder(pigContext)).parse("PigPenEclipsePlugin", cmd,
            // aliasToPlan, opTable));
            lp = ((new LogicalPlanBuilder(pigContext)).parse(
                    "PigPenEclipsePlugin", cmd, aliases, opTable, aliasOp, fileNameMap));
            // lo = opTable.get(lp.getLeaves().get(0));
            lo = lp.getLeaves().get(0);
            // lo = lp.getRootOperator();
        } catch (ParseException e) {
            // throw (IOException) new IOException(e.getMessage()).initCause(e);
            lp = null;
            error = e.getMessage();
        }

        if (lp != null && lo.getAlias() != null) {
            aliasToPlan.put(lo.getAlias(), lp);
        }

        OperatorInfo opInfo = doSimpleParsing(cmd);
        if (opInfo != null) {
            alias = opInfo.outputAlias;
            opName = opInfo.operatorName;
        }

        if (lo instanceof LOStore) {
            // call the processStore method and return
            // FileSpec output = ((LOStore)lo).getOutputFileSpec();
            processStore(cmd, (LOStore) lo);
            return;
        }

        if (lp != null && !opName.equals("Store")) {
            try {
                schema = lo.getSchema();
            } catch (FrontendException e) {
                // TODO Auto-generated catch block
                e.printStackTrace();
            }
            // schema = lo.outputSchema();

        }

        Operator op;
        if (reParseOp == null) { // parsing from scratch
            op = new Operator();
        } else { // re-parsing
            op = reParseOp;
            unplugOp(op, alias, schema);
        }

        Map<LogicalOperator, DataBag> derivedData = null;
        if (lp != null && !opName.equals("Store")) {
            // We don't need to run the example generator here
            // PigExampleGeneratorLauncher ExGen = new
            // PigExampleGeneratorLauncher(lp);
            // Thread t = new Thread(ExGen);
            // //logger.setFocus();
            // //logger.info("Evaluating examples. Please wait...");
            // t.start();
            // try {
            // t.join();
            // } catch (InterruptedException e) {
            // // TODO Auto-generated catch block
            // logger.error(e.getMessage());
            // }
            //			
            // derivedData = ExGen.derivedData;
            // aliasToOp.put(alias, op);
            // aliasToLogOp.put(alias, lp.getRootOperator());

            opToLogOp.put(op, lo);
            LogOpToOp.put(lo, op);

        }
        // System.out.println("Example data generated = " + exdata);

        op.setText(cmd);
        op.setError(error);

        op.setName(opName);
        op.setAlias(alias);

        op.setPlan(lp);
        op.setSchema(schema);

        if (reParseOp == null)
            query.addOperator(op);

        if (lp != null) {
            connectOpInputs(op, lo);
        } else {
            // if unable to resolve this operator's inputs, add to end of
            // reparse queue
            toReparse.add(op);
        }

        // update example data of the entire plan
        // if(lp != null && derivedData != null && error == null)
        // updatePlanExdata(op, derivedData);
    }

    private void updatePlanExdata(Operator op,
            Map<LogicalOperator, DataBag> derivedData) {
        /*
         * String alias = op.getAlias();
         * op.setExdata(derivedData.get(aliasToLogOp.get(alias)).toString());
         */

        op.setExdata(derivedData.get(opToLogOp.get(op)).toString());

        for (Link input : op.getInputs()) {
            Operator source = input.getSource();
            updatePlanExdata(source, derivedData);
        }
    }

    private void unplugOp(Operator op, String newAlias, Schema newSchema) {
        op.removeAllInputs();
        removeOpState(op);

        // check whether downstream operators affected by change in this
        // operator
        // (downstream operators affected if alias or schema of this operator
        // changed)
        String oldAlias = op.getAlias();
        Schema oldSchema = op.getSchema();
        if (oldAlias == null || newAlias == null
                || (!oldAlias.equals(newAlias)) || oldSchema == null
                || newSchema == null || (!oldSchema.equals(newSchema))) {
            // TODO: implement schema.equals() to do semantic equality check, so
            // above comparison actually works

            for (Link output : op.getOutputs())
                toReparse.add(output.getTarget());
        }
    }

    private OperatorInfo doSimpleParsing(String cmd) {
        // use basic parsing to get alias + op name
        InputStream stream = new ByteArrayInputStream((cmd + "\n").getBytes());
        SimpleQueryParser simpleParser = new SimpleQueryParser(stream);
        OperatorInfo opInfo = null;
        try {
            opInfo = simpleParser.Parse();
        } catch (Throwable e) {
            // ignore
        }

        try {
            stream.close();
        } catch (IOException e) {
            // just ignore
        }

        return opInfo;
    }

    private void connectOpInputs(Operator op, LogicalOperator logOp)
            throws IOException {

        if (logOp instanceof LOLoad)
            return;
        for (LogicalOperator lOp : logOp.getPlan().getPredecessors(logOp)) {
            Operator source = LogOpToOp.get(lOp);
            query.addLink(source, op);
        }

    }

    protected void processRemove(String path) throws IOException {
        // no-op
        LoggerUtil.logInfo("Ignoring rm command.");
    }

    @Override
    protected void processCD(String path) throws IOException {
        // TODO: keep track of the current directory,
        // and rewrite all relative file names to be in terms of the current
        // directory
        throw new IOException("CD command currently not supported in PigPen.");
    }

    @Override
    protected void processCopy(String src, String dst) throws IOException {
        // no-op
        LoggerUtil.logInfo("Ignoring copy command.");
    }

    @Override
    protected void processCopyFromLocal(String src, String dst)
            throws IOException {
        // no-op
        LoggerUtil.logInfo("Ignoring copyFromLocal command.");
    }

    @Override
    protected void processCopyToLocal(String src, String dst)
            throws IOException {
        // no-op
        LoggerUtil.logInfo("Ignoring copyToLocal command.");
    }

    @Override
    protected void processDescribe(String alias) throws IOException {
        // no-op
        LoggerUtil.logInfo("Ignoring describe command.");
    }

    @Override
    protected void processMkdir(String dir) throws IOException {
        // no-op
        LoggerUtil.logInfo("Ignoring mkdir command.");
    }

    @Override
    protected void processPWD() throws IOException {
        // no-op
        LoggerUtil.logInfo("Ignoring pwd command.");
    }

    @Override
    protected void processRegister(String jar) throws IOException {
        query.addJar(jar);
        // pigServer.registerJar(jar);
        pigContext.addJar(jar);
    }

    /*
     * @Override protected void processRegisterFunc(String name, String expr) {
     * query.addFunc(name, expr); }
     */

    @Override
    protected void processSet(String key, String value) throws IOException,
            org.apache.pig.tools.pigscript.parser.ParseException {
        query.addSetting(key, value);
    }

    /* @Override */
    protected void processStore(String cmd, LOStore lo) throws IOException {
        FileSpec output = lo.getOutputFile();
        String file = output.getFileName();
        String func = output.getFuncName();
        String exdata = null;

        Operator op;
        if (reParseOp == null) { // parsing from scratch
            op = new Operator();
        } else { // re-parsing one operator
            op = reParseOp;
            unplugOp(op, null, null);
        }

        /*
         * String cmd = "store " + alias + " into '" + file + "'"; if (func !=
         * null && func.length() > 0) cmd = cmd + " using " + func; cmd = cmd +
         * ";"; //String cmd = "Store"; String error = null; if
         * (!aliasToOp.containsKey(alias)) error = "Unrecognized alias: " +
         * alias;
         */

        String error = null;

        op.setName("Store");
        op.setText(cmd);
        op.setError(error);
        op.setExdata(exdata);
        op.setAlias(null);
        // op.setIR(null);
        op.setPlan(null);
        op.setSchema(null);

        if (reParseOp == null)
            query.addOperator(op);

        /*
         * Operator child = aliasToOp.get(alias); if (child != null) {
         * query.addLink(child, op); } else { // if unable to resolve this
         * operator's inputs, add to end of reparse queue toReparse.add(op); }
         */
        if (lo != null) {
            connectOpInputs(op, lo);
        } else {
            // if unable to resolve this operator's inputs, add to end of
            // reparse queue
            toReparse.add(op);// update example data of the entire plan

        }
    }

    @Override
    public void prompt() {
        // don't print a prompt
    }

    @Override
    protected void quit() {
        mDone = true;
    }



    @Override
    protected void processIllustrate(String arg0) throws IOException {
        // TODO Auto-generated method stub

    }

    class PigExampleGeneratorLauncher implements Runnable {

        LogicalPlan lp;
        String Result;
        Map<LogicalOperator, DataBag> derivedData;

        public PigExampleGeneratorLauncher(LogicalPlan lp) {
            this.lp = lp;
            // this.Result = result;
        }

        public void run() {
            // try {
            // this.derivedData = ExGen.GenerateExamples(lp, pigContext);
            // } catch (IOException e) {
            // // TODO Auto-generated catch block
            // e.printStackTrace();
            // }
            this.derivedData = pigServer.getExamples(lp.getLeaves().get(0)
                    .getAlias());

        }

    }

	@Override
	protected void printAliases() throws IOException {
		// TODO Auto-generated method stub
		
	}

	@Override
	protected void processExplain(String arg0, String arg1, boolean arg2,
			String arg3, String arg4, List<String> arg5, List<String> arg6)
			throws IOException,
			org.apache.pig.tools.pigscript.parser.ParseException {
		// TODO Auto-generated method stub
		
	}

	@Override
	protected void processRemove(String arg0, String arg1) throws IOException {
		// TODO Auto-generated method stub
		
	}

	@Override
	protected void processScript(String arg0, boolean arg1, List<String> arg2,
			List<String> arg3) throws IOException,
			org.apache.pig.tools.pigscript.parser.ParseException {
		// TODO Auto-generated method stub
		
	}
}
